package cn.dapeng.rpc.client.pool;

import cn.dapeng.rpc.common.codec.ClientByteToMessageDecoder;
import cn.dapeng.rpc.common.handler.ClientResponseHandler;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;

import java.net.InetSocketAddress;
import java.util.Random;
import java.util.concurrent.ConcurrentHashMap;

public class ClientFactoryV2 {

    private final int POOL_SIZE = 1000;

    private Random random = new Random();

    static {
        instance = new ClientFactoryV2();
    }

    private NioEventLoopGroup clientWorker;
    private static ClientFactoryV2 instance;

    private ClientFactoryV2() {
    }

    public static ClientFactoryV2 getInstance() {
        return instance;
    }

    private ConcurrentHashMap<InetSocketAddress, ClientPool> outBox = new ConcurrentHashMap<>();

    public NioSocketChannel getClient(InetSocketAddress address) {
        ClientPool clientPool = outBox.get(address);

        if (clientPool == null) {
            synchronized (outBox) {
                if (clientPool == outBox.get(address)) {
                    outBox.putIfAbsent(address, new ClientPool(POOL_SIZE));
                    clientPool = outBox.get(address);
                }
            }
        }

        int i = random.nextInt(POOL_SIZE);
        if (clientPool.client[i] != null && clientPool.client[i].isActive()) {
            return clientPool.client[i];
        }

        synchronized (clientPool.monitor[i]) {
            if (clientPool.client[i] == null || !clientPool.client[i].isActive()) {
                clientPool.client[i] = create(address);
            }
        }
        return clientPool.client[i];
    }

    private NioSocketChannel create(InetSocketAddress address) {
        clientWorker = new NioEventLoopGroup(1);
        Bootstrap bootstrap = new Bootstrap();
        ChannelFuture connect = bootstrap.group(clientWorker).channel(NioSocketChannel.class).handler(new ChannelInitializer<NioSocketChannel>() {
            @Override
            protected void initChannel(NioSocketChannel channel) throws Exception {
                channel.pipeline()
                        .addLast(new ClientByteToMessageDecoder())
                        .addLast(new ClientResponseHandler());
            }
        }).connect(address);

        try {
            return (NioSocketChannel) connect.sync().channel();
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }

}
